use std::{
    sync::{atomic::AtomicBool, Arc},
    time::Duration,
};

use node::NodeSocket;
use serde::Deserialize;
use shutdown::{Shutdown, ShutdownSignal};
use tokio::net::TcpStream;

use crate::{
    client::RedisOptions,
    commands::{self, HelloResponse},
    resp::{Command, Value},
    RedisError, RedisResult,
};

pub(crate) mod cluster;
mod node;
pub(crate) mod shutdown;
pub(crate) mod standalone;

const RETRY_INTERVAL: Duration = Duration::from_millis(600);

pub enum Connection {
    Standalone(standalone::StandaloneConnection),
    Cluster(cluster::ClusterConnection),
    #[allow(dead_code)]
    Sentinel,
}

#[repr(u8)]
pub enum ConnectionStatus {
    Initialize = 0,
    Connecting = 1,
    Connected = 2,
    Disconnected = 3,
    Reconnecting = 4,
}

// impl ConnectionStatus {
//     pub fn is_connected(&self) -> bool {
//         match self {
//             ConnectionStatus::Connected => true,
//             _ => false,
//         }
//     }
// }

impl From<u8> for ConnectionStatus {
    fn from(value: u8) -> Self {
        match value {
            0 => ConnectionStatus::Initialize,
            1 => ConnectionStatus::Connecting,
            2 => ConnectionStatus::Connected,
            3 => ConnectionStatus::Disconnected,
            4 => ConnectionStatus::Reconnecting,
            _ => ConnectionStatus::Initialize,
            // _ => panic!("invalid connectionStatus value"),
        }
    }
}

pub(crate) struct ConnectionState {
    pub(crate) client_name: Option<String>,
    pub(crate) connect_timeout: Duration,
    pub(crate) reconnect_interval: Duration,
    pub(crate) shutdown: Shutdown,
    pub(crate) receive_timeout: Duration,
    pub(crate) user: Option<String>,
    pub(crate) password: Option<String>,
    pub(crate) stream_tx: tokio::sync::Mutex<Option<tokio::sync::mpsc::Sender<Value>>>,
    pub(crate) stream_flag: AtomicBool,
    pub(crate) timeout: Option<Duration>,
    pub(crate) retry_on_send_error: u8,
}

impl ConnectionState {
    pub(crate) fn new(options: RedisOptions) -> Self {
        Self {
            client_name: options.client_name,
            connect_timeout: options.connect_timeout,
            reconnect_interval: options.reconnect_interval,
            shutdown: Shutdown::new(),
            receive_timeout: options.receive_timeout,
            user: options.user,
            password: options.password,
            stream_tx: tokio::sync::Mutex::new(None),
            stream_flag: AtomicBool::new(false),
            timeout: options.timeout,
            retry_on_send_error: options.retry_on_send_error,
        }
    }
    // pub(crate) fn new<S: Into<FastStr>>(
    //     host: S,
    //     ip: u32,
    //     connect_timeout: Duration,
    //     reconnect_interval: Duration,
    // ) -> Self {
    //     Self {
    //         host: host.into(),
    //         ip,
    //         connect_timeout,
    //         reconnect_interval,
    //         shutdown_sender: tokio::sync::broadcast::channel::<()>(3).0,
    //     }
    // }

    pub(crate) fn shutdown_tx_clone(&self) -> tokio::sync::mpsc::Sender<ShutdownSignal> {
        self.shutdown.clone_tx()
    }

    pub(crate) async fn enter_stream_mode(&self) -> tokio::sync::mpsc::Receiver<Value> {
        let (tx, rx) = tokio::sync::mpsc::channel::<Value>(5);
        {
            let mut guard = self.stream_tx.lock().await;
            guard.replace(tx);
        }
        self.stream_flag
            .store(true, std::sync::atomic::Ordering::Release);
        rx
    }

    pub(crate) async fn exit_stream_mode(&self) {
        let mut guard = self.stream_tx.lock().await;
        guard.take();
        self.stream_flag
            .store(false, std::sync::atomic::Ordering::Release);
    }
}

async fn reconnect(epoch: u64, state: Arc<ConnectionState>, socket: Arc<NodeSocket>) {
    if !socket.epoch_compare_advance(epoch) {
        return;
    }
    socket.set_disconnected().await;
    state.exit_stream_mode().await;
    tokio::spawn(start_reconnect(state, socket));
}

async fn start_reconnect(state: Arc<ConnectionState>, socket: Arc<NodeSocket>) {
    let addr = format!("{}:{}", socket.host, socket.port);
    log::info!("try reconnect to {addr}, epoch = {}", socket.epcoh_get());
    socket.set_reconnecting();
    loop {
        match internal_connect(&state, &socket, &addr).await {
            Ok(_) => {
                break;
            }
            Err(err) => {
                log::error!("{err}");
            }
        }
        tokio::time::sleep(state.reconnect_interval).await;
    }
}

async fn internal_connect(
    state: &Arc<ConnectionState>,
    socket: &Arc<NodeSocket>,
    addr: &str,
) -> RedisResult<()> {
    match tokio::time::timeout(state.connect_timeout, TcpStream::connect(&addr)).await {
        Ok(result) => match result {
            Ok(stream) => internal_on_connected(state, socket, stream).await,
            Err(err) => Err(RedisError::new_connect(
                format!("error on connect to {addr}, epoch = {}", socket.epcoh_get()),
                err,
            )),
        },
        Err(_err) => Err(RedisError::Connect(format!(
            "error on connect to {addr}, epoch = {}, timeout in {}s",
            socket.epcoh_get(),
            state.connect_timeout.as_secs()
        ))),
    }
}

async fn internal_on_connected(
    state: &Arc<ConnectionState>,
    socket: &Arc<NodeSocket>,
    stream: TcpStream,
) -> RedisResult<()> {
    socket.epoch_advance();
    _ = stream.nodelay();
    let (read_half, write_half) = stream.into_split();
    node::start_stream_read(
        state.clone(),
        socket.clone(),
        read_half,
        state.shutdown_tx_clone(),
    );
    socket.set_connected(write_half).await;
    log::trace!("connected, epoch = {}", socket.epcoh_get());
    let response = internal_hello(state, socket).await?;
    // if state.mode != response.mode {

    // }
    log::trace!("receive {:?}", response);
    Ok(())
}

async fn internal_hello(
    state: &ConnectionState,
    socket: &NodeSocket,
) -> RedisResult<HelloResponse> {
    let cmd_hello = commands::build_hello_command(state);
    internal_send_command::<HelloResponse>(state, socket, cmd_hello).await
}

#[inline]
async fn internal_send_command<'a, T: Deserialize<'a>>(
    state: &ConnectionState,
    socket: &NodeSocket,
    command: Command<'a>,
) -> RedisResult<T> {
    let bytes = command.encode();
    let rx = match socket.send(&bytes).await {
        Ok(rx) => rx,
        Err(err) => {
            log::error!("{err:?}");
            internal_send_command_retry(state, socket, &bytes).await?
        }
    };
    // let rx = socket.send(&bytes).await?;
    let val = wait_receive_after_send(state, rx).await?;
    val.read_error()?;
    T::deserialize(&val)
}

async fn internal_send_command_retry(
    state: &ConnectionState,
    socket: &NodeSocket,
    content: &[u8],
) -> RedisResult<tokio::sync::oneshot::Receiver<Value>> {
    for i in 0..state.retry_on_send_error {
        tokio::time::sleep(Duration::from_millis(600)).await;
        if i == state.retry_on_send_error - 1 {
            return socket.send(&content).await;
        } else {
            match socket.send(&content).await {
                Ok(val) => {
                    return Ok(val);
                }
                Err(err) => {
                    log::error!("error on retry {}, {err}", i + 1);
                }
            }
        }
    }
    Err(RedisError::Other("error on send_command_retry".to_string()))
}

/// receive the value from the response of redis server after send command
async fn wait_receive_after_send(
    state: &ConnectionState,
    rx: tokio::sync::oneshot::Receiver<Value>,
) -> RedisResult<Value> {
    match tokio::time::timeout(state.receive_timeout, rx).await {
        Ok(res) => {
            match res {
                Ok(val) => {
                    Ok(val)
                }
                Err(err) => Err(RedisError::new_wait_receive("error on receive result", err)),
            }
        }
        Err(_err) => Err(RedisError::WaitReceive(format!(
            "timeout in {}s",
            state.receive_timeout.as_secs()
        ))),
    }
}
